package com.pangyangyang.elasticsearch.plugin.dispatch;

import com.pangyangyang.elasticsearch.plugin.event.DataModifyEvent;
import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.handler.EventHandler;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 注册消息处理器
 * 负责接收并分发消息
 * Created by yangwang on 2015/1/15.
 */
public class EventDispatcher {

    private Map<String, EventHandler> registeredHandlers = new HashMap<>();
    private ExecutorService executorService;
    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    public EventDispatcher() {
        executorService = Executors.newFixedThreadPool(1);
    }

    public Future dispatchEvent(DataModifyEvent event){
        String destination = event.getDataDestination();
        EventHandler handler = registeredHandlers.get(destination);
        EventDispatchTask task = new EventDispatchTask(event, handler);
        logger.debug("submit dispatch task des {}", destination);

        Future future = executorService.submit(task);
        return future;
    }

    public synchronized void registerHandler( EventHandler handler) throws MessageRiverException{

        String destination = handler.getDestination();
        if(registeredHandlers.containsKey(destination)){
            throw new MessageRiverException(String.format("Destination %s already registered.", destination));
        }
        registeredHandlers.put(destination, handler);
    }

    public void unregisterHandler(String destination){
        registeredHandlers.remove(destination);
    }

    public void stop(){
        executorService.shutdown();
    }
}
